package com.jungle.task;

import com.jungle.task.executor.*;
import com.jungle.task.parameter.*;
import com.jungle.task.result.*;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author 李岩 at 2021/10/22.
 */
@Slf4j
public class TaskDispatcher {
    private final ConvertTaskExecutor convertTaskExecutor;
    private final List<InsertTaskExecutor> insertTaskExecutor;
    private final List<QueryTaskExecutor> queryTaskExecutor;

    public TaskDispatcher(final List<TaskExecutor<?,?>> executors){
        this.convertTaskExecutor =
            (ConvertTaskExecutor)filter(executors,predicate(TaskType.CONVERT)).findFirst().orElseGet(
                DefaultConvertTaskExecutor::new);
        this.insertTaskExecutor =
            filter(executors,predicate(TaskType.INSERT)).map(InsertTaskExecutor.class::cast).collect(Collectors.toList());
        this.queryTaskExecutor =
            filter(executors,predicate(TaskType.QUERY)).map(QueryTaskExecutor.class::cast).collect(Collectors.toList());
    }

    public ConvertTaskResult dispatchConvertTask(final Task<ConvertTaskParameter, ConvertTaskResult> task){
        ConvertTaskResult execute = null;
        try {
            execute = this.convertTaskExecutor.execute(task);
        }catch (final Exception e){
            log.error("fsc plugin，执行convert task，执行失败，task:{},e:",task,e);
        }
        task.setResult(Collections.singletonMap(this.convertTaskExecutor,execute));
        return execute;
    }


    public void dispatchSaveTask(final Task<SaveTaskParameter, SaveTaskResult> task){
        dispatchSaveTask(task,this.insertTaskExecutor);
    }

    public void dispatchSaveTask(final Task<SaveTaskParameter, SaveTaskResult> task, final List<InsertTaskExecutor> executors){
        for (final InsertTaskExecutor taskExecutor : executors) {
            SaveTaskResult execute = null;
            try {
                execute = taskExecutor.execute(task);
            }catch (final Exception e){
                log.error("fsc plugin，执行save task，执行失败，task:{},e:",task,e);
            }
            task.getResult().put(taskExecutor, execute);
        }
    }

    public QueryTaskResult dispatchQueryTask(final Task<QueryTaskParameter, QueryTaskResult> task){
        QueryTaskResult result = null;
        for (final QueryTaskExecutor taskExecutor : this.queryTaskExecutor) {
            try {
                result = taskExecutor.execute(task);
                if (Objects.nonNull(result) && Objects.nonNull(result.getOriginal())) {
                    break;
                }
            }catch (final Exception e){
                log.error("fsc plugin，执行query task，执行失败，task:{},e:",task,e);
            }
        }
        return result;
    }

    public CompensateTaskResult dispatchCompensateTask(final Task<CompensateTaskParameter, CompensateTaskResult> task){
        //do nothing, let sub class impl
        return null;
    }

    protected Stream<TaskExecutor<? extends TaskParameter,? extends TaskResult>> filter(final List<TaskExecutor<?,?>> executors,
        final Predicate<TaskExecutor<?,?>> predicate){
        return Optional.ofNullable(executors)
            .orElseGet(Collections::emptyList)
            .stream()
            .filter(predicate);
    }

    protected Predicate<TaskExecutor<? extends TaskParameter,? extends TaskResult>> predicate(final TaskType taskType){
        return f -> Objects.equals(taskType,f.concernType());
    }
}
